package main

import (
	"context"
	"errors"
	"github.com/segmentio/kafka-go"
	"log"
	"time"
)

func main() {
	w := kafka.Writer{
		Addr:                   kafka.TCP("localhost:9092"),
		Topic:                  "test-mch",
		Balancer:               &kafka.Hash{},
		AllowAutoTopicCreation: false,
	}
	msg := []kafka.Message{
		{Key: []byte("1"), Value: []byte(`-你好哦???`)},
		{Key: []byte("2"), Value: []byte(`-kfaxxx!!!`)},
		{Key: []byte("3"), Value: []byte(`-你好哦333@@@@`)},
		{Key: []byte("100"), Value: []byte(`-你好哦333@@@@`)},
	}

	for {
		ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(5*time.Second))
		defer cancel()
		err := w.WriteMessages(ctx, msg...)
		if errors.Is(err, kafka.LeaderNotAvailable) || errors.Is(err, context.DeadlineExceeded) {
			continue
		}
		if err != nil {
			log.Fatal(err)
		}
		break
	}
	if err := w.Close(); err != nil {
		log.Fatal(err)
	}
}
